package com.atguigu.realtime.sink;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.realtime.bean.TableProcess;
import com.atguigu.realtime.util.JdbcUtil;
import com.atguigu.realtime.util.RedisUtil;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import redis.clients.jedis.Jedis;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Map;
import java.util.Set;

/**
 * @Author lzc
 * @Date 2022/7/18 9:17
 */
public class PhoenixSink extends RichSinkFunction<Tuple2<JSONObject, TableProcess>> {
    
    private Connection conn;
    private ValueState<String> sqlState;
    private Jedis jedis;
    
    @Override
    public void open(Configuration parameters) throws Exception {
        //        DruidDataSource dataSource = DruidDSUtil.createDataSource();
        //        conn = dataSource.getConnection();
        conn = JdbcUtil.getPhoenixConnection();
        
        sqlState = getRuntimeContext().getState(new ValueStateDescriptor<String>("sqlState", String.class));
        
        jedis = RedisUtil.getRedisClient();
        
    }
    
    @Override
    public void close() throws Exception {
        if (conn != null) {
            conn.close();  // 连接是从连接池获取的, 所以这里的close不是关闭, 而是归还
        }
        
        if (jedis != null) {
            jedis.close();
        }
    }
    
    @Override
    public void invoke(Tuple2<JSONObject, TableProcess> t,
                       Context ctx) throws Exception {
        
        JSONObject data = t.f0;
        TableProcess tp = t.f1;
        // 1. 把维度数据写入到Phoenix中
        saveDimToPhoenix(data, tp);
        
        // 2.更新(删除) redis中的缓存
        //delCache(data, tp);
        updateCache(data, tp);
        
    }
    
    private void updateCache(JSONObject data, TableProcess tp) {
        String key = tp.getSinkTable() + ":" + data.getString("id");
        // 必须update, 并且缓存中存在才要更新.
        if ("update".equals(data.getString("operate_type")) && jedis.exists(key)){
            data.remove("operate_type");
            // 更新的本质就把维度数据重写一下
            //data只的key都是小写, 所以需要data中的小写的key'变成大写
            JSONObject outer = new JSONObject();
            Set<Map.Entry<String, Object>> set = data.entrySet();
            for (Map.Entry<String, Object> entry : set) {
                outer.put(entry.getKey().toUpperCase(), entry.getValue());
            }
            jedis.setex(key,2 * 24 * 60 * 60, outer.toJSONString());
    
        }
    }
    
    private void delCache(JSONObject data, TableProcess tp) {
        
        // table + id
        // 如果是更新, 才去删, 如果是新增就不需要删除
        // 即使是更新, 如果redis中没有这个缓存要不要删?
        // 可以删, 如果存在就删除成功, 不存在就什么都没有做
        if ("update".equals(data.getString("operate_type"))) {
            String key = tp.getSinkTable() + ":" + data.getString("id");
            jedis.del(key);
        }
    }
    
    private void saveDimToPhoenix(JSONObject data, TableProcess tp) throws java.io.IOException, SQLException {
        // upsert into t(a, b, c)values(?,?,?)
        if (sqlState.value() == null) {
            StringBuilder sql = new StringBuilder();
            sql
                .append("upsert into ")
                .append(tp.getSinkTable())
                .append("(")
                .append(tp.getSinkColumns())  // id,b,c
                .append(")values(")
                .append(tp.getSinkColumns().replaceAll("[^,]+", "?"))
                .append(")");
            sqlState.update(sql.toString());
        }
        
        String sql = sqlState.value();
        System.out.println("插入语句: " + sql);
        PreparedStatement ps = conn.prepareStatement(sql);
        // 给sql中的占位符赋值 TODO
        String[] columns = tp.getSinkColumns().split(",");
        for (int i = 0; i < columns.length; i++) {
            String name = columns[i];
            String v = data.getString(name);
            
            ps.setString(i + 1, v);
        }
        // 执行
        ps.execute();
        conn.commit();
        ps.close();
    }
}
/*
 
 */